package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo4Student {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .master("local")
      .appName("student")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
      *
      * 1、统计班级级排名前十学生各科的分数 [学号,学生姓名，学生班级，科目，分数]
      */

    val score: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("sid STRING , cId STRING,sco INT")
      .load("data/score.txt")

    val student: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING, name STRING , age INT , gender  STRING , clazz STRING")
      .load("data/students.txt")

    val cource: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("cId STRING, cName STRING , maxSco INT")
      .load("data/cource.txt")


    score
      .join(student, $"sid" === $"id")
      .groupBy($"id", $"name", $"clazz")
      //计算学生总分
      .agg(sum($"sco") as "sumSco")
      //多美歌班级所有的学生总分排名
      .withColumn("rank", row_number() over Window.partitionBy($"clazz").orderBy($"sumSco".desc))
      //取前十
      .where($"rank" <= 10)
      //关联分数表，获取各科的分数
      .join(score, $"id" === $"sid")
      //关联科目表
      .join(cource, "cId")
      //取出字段
      .select($"id", $"name", $"clazz", $"cName", $"sco")
      //保存数据
      .write
      .mode(SaveMode.Overwrite)
      .json("data/top10")

    /**
      * 使用sql
      *
      */

    score.createOrReplaceTempView("score")
    student.createOrReplaceTempView("student")
    cource.createOrReplaceTempView("cource")


    val resultDF: DataFrame = spark.sql(
      """
        |select id,name,clazz,cName,sco from (
        |select id,name,clazz,sumSco,row_number() over(partition by clazz order by sumSco desc) as rank from (
        |select id,name,clazz,sum(sco) as sumSco from
        |student as a
        |join
        |score as b
        |on a.id=b.sId
        |group by id,name,clazz
        |)as c
        |)as d
        |join score as e
        |on d.id=e.sId
        |join
        |cource as f
        |on e.cId = f.cId
        |where rank <=10

        |
      """.stripMargin)
    resultDF.show()

  }

}
